/*
 * Copyright (c) 2016 moondev.cn
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package cn.moondev.moonlog.delegate.mq;

import cn.moondev.moonlog.delegate.utils.Identities;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<BR>
 * 请注意：此Producer采用<code>ONEWAY</code>方式发送消息到mq，不保证消息的可靠性，请谨慎使用
 * @author scchen
 */
public class Producer {
	
	private static final Logger log = LoggerFactory.getLogger(Producer.class);
	
	/**
	 * 发送mq失败的消息,单独存放在一个文件中
	 */
	private static final Logger failLog = LoggerFactory.getLogger("mq.producer.fail");
	
	/**
	 * 消息topic，建议一个应用一个topic
	 */
	private String topic;
	
	private DefaultMQProducer producer;
	
	/**
	 * 构造Producer
	 * @param producerGroupName
	 * 				ProducerGroupName需要由应用来保证唯一<BR>
	 * 				ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，因为服务器会回查这个Group下的任意一个Producer
	 * @param namesrvAddr
	 * 				name srv 的ip地址以及host，格式：ip:port
	 * @param instanceName
	 * @param topic
	 * 				消息topic，建议一个应用一个topic
	 */
	public Producer(String producerGroupName, String namesrvAddr, String instanceName, String topic) {
		log.info("开始实例化Producer,producerGroupName={},namesrvAddr={},instanceName={},默认topic={}",producerGroupName,namesrvAddr,instanceName,topic);
		this.topic = topic;
		this.producer = new DefaultMQProducer(producerGroupName);
		this.producer.setNamesrvAddr(namesrvAddr);
		this.producer.setInstanceName(instanceName);
	}
	
	/**
	 * 发送消息
	 * @param tag
	 * @param text
	 */
	public void send(String tag,String text) {
		send(tag,text,topic);
	}
	
	/**
	 * 发送消息
	 * @param tag	消息TAG，建议一种类型的消息用同一个TAG
	 * @param text	消息正文
	 */
	public void send(String tag,String text,String topic) {
		log.info("开始发送mq消息：tag={},text={},topic={}",tag,text,topic);
		Message msg = new Message(topic, tag, Identities.uuid2(),text.getBytes());
		try {
			/**
			 * 发送状态分为4种：
			 * SEND_OK:消息发送成功
			 * FLUSH_DISK_TIMEOUT:消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
			 * FLUSH_SLAVE_TIMEOUT:消息发送成功，但是服务器同步到 Slave 时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
			 * SLAVE_NOT_AVAILABLE:消息发送成功，但是此时 slave 不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢
			 */
			// 注意：日志采集应用不要求高的可靠性，所以不管发送成功与否，均不再处理
			producer.sendOneway(msg);
		} catch (Exception e) {
			failLog.error("rocketmq producer send message error:tag={},text={},topic={}",tag,text,topic,e);
		}
	}
	
	public void start() {
		if (producer != null) {
			try {
				log.info("执行producer.start方法");
				producer.start();
			} catch (MQClientException e) {
				failLog.error("rocketmq producer start error",e);
			}
		}
	}
	
	public void shutdown() {
		if (producer != null) {
			log.info("执行producer.shutdown方法");
			producer.shutdown();
		}
	}
}
